package com.shadow.disnet.disrutor;


import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.util.concurrent.DefaultThreadFactory;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DisruptorExecutor implements Executor {

    private Disruptor<Event> disruptor;
    private RingBuffer<Event> ringBuffer;
    private ExecutorService executorService;

    public DisruptorExecutor(int bufferSize, String threadName) {
        if (!checkSize(bufferSize)) {
            throw new RuntimeException("buffSize is pow of 2");
        }
        DefaultThreadFactory threadFactory = new DefaultThreadFactory(threadName);
        this.executorService = Executors.newFixedThreadPool(10,threadFactory);
        EventFactory<Event> eventFactory = new ShadowEventFactory();
        this.disruptor = new Disruptor<>(eventFactory, bufferSize, executorService, ProducerType.MULTI, new BlockingWaitStrategy());
        this.disruptor.handleEventsWith(new ShadowEventHandler());
        this.ringBuffer = disruptor.start();
    }

    private static boolean checkSize(int size) {
        return size > 0 && (size & size - 1) == 0;
    }

    @Override
    public void execute(Runnable task) {
        long next = ringBuffer.next();
        try {
            Event event = ringBuffer.get(next);
            event.setTask(task); //将事件填充进去
        } finally {
            ringBuffer.publish(next);
        }
    }

    public void processMessage(int moduleId, int msgId, String byteBuf, MessageProducer messageProducer) {
        long next = ringBuffer.next();
        try {
            Event event = ringBuffer.get(next);
            event.setEvent(moduleId, msgId, byteBuf, messageProducer); //将消息封装成事件
        } finally {
            ringBuffer.publish(next);
        }
    }
}
